package dubbo.learn.transport.mina;

import dubbo.learn.common.ProviderHostAndPort;
import org.apache.dubbo.common.utils.NamedThreadFactory;
import org.apache.dubbo.common.utils.NetUtils;
import org.apache.dubbo.remoting.Constants;
import org.apache.mina.common.*;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.transport.socket.nio.SocketConnector;
import org.apache.mina.transport.socket.nio.SocketConnectorConfig;

import java.net.InetSocketAddress;
import java.util.concurrent.Executors;

/**
 * Mina client.
 */
public class CustomMinaClient {

    private ProviderHostAndPort providerHostAndPort;


    private SocketConnector connector;

    private volatile IoSession session; // volatile, please copy reference to use

    MinaLengthFieldPrepender encoder = new MinaLengthFieldPrepender();
    MyDataDecoder decoder = new MyDataDecoder();

    public CustomMinaClient(ProviderHostAndPort providerHostAndPort) {
        this.providerHostAndPort = providerHostAndPort;
        connector = new SocketConnector(Constants.DEFAULT_IO_THREADS,
                Executors.newCachedThreadPool(new NamedThreadFactory("MinaClientWorker", true)));
        // 配置参数
        SocketConnectorConfig cfg = (SocketConnectorConfig) connector.getDefaultConfig();
        cfg.setThreadModel(ThreadModel.MANUAL);
        cfg.getSessionConfig().setTcpNoDelay(true);
        cfg.getSessionConfig().setKeepAlive(true);
        // 1s超时
        cfg.setConnectTimeout(1);

        // set codec.
        ProtocolCodecFilter filter = new ProtocolCodecFilter(encoder,decoder);
        connector.getFilterChain().addLast("codec", filter);



        InetSocketAddress inetSocketAddress = new InetSocketAddress(
                NetUtils.filterLocalHost(providerHostAndPort.getHost()),
                providerHostAndPort.getPort());
        ConnectFuture future = connector.connect(inetSocketAddress,
                new CustomMinaHandler());
        future.join();
        this.session = future.getSession();
    }


    public void send(Object message) {
        try {
            session.write(message);
        } catch (Throwable e) {
            throw new RuntimeException();
        }

    }
}
